package com.crux.ribbon;

import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang.StringUtils;
import org.springframework.beans.BeansException;
import org.springframework.cloud.client.ServiceInstance;
import org.springframework.cloud.client.loadbalancer.Request;
import org.springframework.cloud.loadbalancer.core.DelegatingServiceInstanceListSupplier;
import org.springframework.cloud.loadbalancer.core.ServiceInstanceListSupplier;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationContextAware;
import reactor.core.publisher.Flux;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicLong;
import java.util.stream.Collectors;

@Slf4j
public class BestAvailableLoadServiceInstanceListSupplier extends DelegatingServiceInstanceListSupplier implements ApplicationContextAware{
    private ServiceInstanceListSupplier delegate;
    private ApplicationContext applicationContext;

    public BestAvailableLoadServiceInstanceListSupplier(ServiceInstanceListSupplier delegate) {
        super(delegate);
        this.delegate = delegate;
//        this.myLoadBalancerLifecycle = myLoadBalancerLifecycle;
    }

    @Override
    public Flux<List<ServiceInstance>> get() {
        return delegate.get().map(this::filterByVersion);
    }

    @Override
    public Flux<List<ServiceInstance>> get(Request request) {
        return delegate.get(request).map(this::filterByVersion);
    }

    private List<ServiceInstance> filterByVersion(List<ServiceInstance> serviceInstances) {
        //当前需要负载的服务名称
        String destServiceId = delegate.getServiceId();
        //从注册中心获取当前负载服务的全部实例
        List<ServiceInstance> eurekaInstances = serviceInstances.stream().filter(serviceInstance -> StringUtils.equalsIgnoreCase(destServiceId, serviceInstance.getServiceId())).collect(Collectors.toList());
        //获取生命周期的示例
        CruxLoadBalancerLifecycle cruxLoadBalancerLifecycle = applicationContext.getBean(CruxLoadBalancerLifecycle.class);
        ConcurrentHashMap<ServiceInstance, AtomicLong> activeRequestsPerInstance = cruxLoadBalancerLifecycle.getActiveRequestsPerInstanceUsed();
        //过滤当前服务在activeRequestsPerInstance的集合
        Map<ServiceInstance,Long> phaseServiceInstance = new HashMap<>();
        for (Map.Entry<ServiceInstance,AtomicLong> entry : activeRequestsPerInstance.entrySet()) {
            if(entry.getKey().getServiceId().equalsIgnoreCase(destServiceId)){
                phaseServiceInstance.put(entry.getKey(),entry.getValue().get());
            }
        }
        if(phaseServiceInstance.size() == 0){
            log.debug("走默认的轮询策略");
        }else if(phaseServiceInstance.size() < eurekaInstances.size()){
            List<String> instanceIds = phaseServiceInstance.keySet().stream().map(ServiceInstance::getInstanceId).collect(Collectors.toList());
            return serviceInstances.stream().filter(serviceInstance -> !instanceIds.contains(serviceInstance.getInstanceId())).collect(Collectors.toList());
        } else if (phaseServiceInstance.size() == eurekaInstances.size()){
            //获取最小值的service实例
            List<String> minInstanceIds = getMinInstanceId(phaseServiceInstance);
            return serviceInstances.stream().filter(serviceInstance -> minInstanceIds.contains(serviceInstance.getInstanceId())).collect(Collectors.toList());
        }else{
            log.debug("生命周期的示例数量不可能大于注册中心的示例数量");
        }
        return eurekaInstances;
    }

    private List<String> getMinInstanceId(Map<ServiceInstance, Long> phaseServiceInstance) {
        List<String> result = new ArrayList<>();
        Long min = phaseServiceInstance.values().stream().min(Long::compare).get();
        for (Map.Entry<ServiceInstance,Long> entry : phaseServiceInstance.entrySet()) {
            if(entry.getValue().longValue() == min.longValue()){
                result.add(entry.getKey().getInstanceId());
            }
        }
        return result;
    }

    @Override
    public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
        this.applicationContext = applicationContext;
    }
}
